package com.smy.baseutil.concurrent;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**   
  * @Description: 并发测试工具类
  * @author zhoutao
  * @date 2017年8月17日 下午2:28:47 
  * @version V1.0   
  */
public class ConcurrentUtil {
	
	/**
	 * 同步栅栏
	 */
	private static CyclicBarrier cb = null;
	
	/**
	 * 并发的线程数
	 * @param threadCount
	 * @return
	 */
	private static Integer getCount(final Integer threadCount){
		return (threadCount<=0||threadCount>Integer.MAX_VALUE) ? 100 : threadCount;
	}
	
	/**
	 * Class模式
	 * @param clazz
	 * @param methodName
	 * @param args
	 * @param threadCount
	 * @return
	 * @throws Exception
	 */
	public static <T> Map<String, Object> exec(final Class<T> clazz, final String methodName, final Object args, final Integer threadCount) throws Exception {
		return exec(threadCount, new BizSubmit() {
			@Override
			public Object submit() throws Exception {
				return invoke(clazz, methodName, args);
			}
		});
	}
	
	/**
	 * Object模式
	 * @param obj
	 * @param methodName
	 * @param args
	 * @param threadCount
	 * @return
	 * @throws Exception
	 */
	public static Map<String, Object> exec(final Object obj, final String methodName, final Object args, final Integer threadCount) throws Exception {
		return exec(threadCount, new BizSubmit() {
			@Override
			public Object submit() throws Exception {
				return invoke(obj, methodName, args);
			}
		});
	}
	
	/**
	 * 基础方法，
	 * 采用CyclicBarrier线程等待，同时执行
	 * @param threadCount
	 * @param bizSubmit
	 * @return
	 * @throws Exception
	 */
	public static <T> Map<String, Object> exec(final Integer threadCount, final BizSubmit bizSubmit) throws Exception {
		final int count = getCount(threadCount);
		cb = new CyclicBarrier(count);
		final Map<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
		for (int i = 0, j = count; i < j; i++) {
			new Thread(new Runnable() {
				@Override
				public void run(){
					try {
						cb.await();
						resultMap.put(Thread.currentThread().getName(), bizSubmit.submit());
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (BrokenBarrierException e) {
						e.printStackTrace();
					} catch (SecurityException e) {
						e.printStackTrace();
					} catch (IllegalArgumentException e) {
						e.printStackTrace();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}).start();
		}
		
		Thread.sleep(1000L);
		
		return resultMap;
	}
	
	/**
     * 基础方法，
     * 采用CyclicBarrier线程等待，同时执行
     * @param threadCount
     * @param bizSubmit
     * @return
     * @throws Exception
     */
    public static <T> Map<String, Object> exec(final Integer threadCount, final List<BizSubmit> bizSubmits) throws Exception {
        final int count = getCount(threadCount);
        final int size = bizSubmits.size();
        cb = new CyclicBarrier(count);
        final Map<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
        for (int i = 0, j = count; i < j; i++) {
            final AtomicInteger num = new AtomicInteger(i);
            new Thread(new Runnable() {
                @Override
                public void run(){
                    try {
                        cb.await();
                        resultMap.put(Thread.currentThread().getName(), bizSubmits.get(num.get()%size).submit());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    } catch (SecurityException e) {
                        e.printStackTrace();
                    } catch (IllegalArgumentException e) {
                        e.printStackTrace();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        
        Thread.sleep(1000L);
        
        return resultMap;
    }

	/**
	 * 线程池模式的并发，
	 * 底层还是一个一个的执行的，伪并发，弱并发
	 * @param obj
	 * @param methodName
	 * @param args
	 * @param threadCount
	 * @return
	 * @throws Exception
	 */
	public static Set<Object> execWeak(final Object obj, final String methodName, final Object args, final Integer threadCount) throws Exception {
		final Set<Object> set = Collections.synchronizedSet(new HashSet<Object>());
		
		final int count = getCount(threadCount);
		ThreadPoolExecutor tpe = new ThreadPoolExecutor(count, count, 2000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(count));
		
		List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();
		for (int i = 0; i < count; i++) {
			tasks.add(new Callable<Object>() {
				@Override
				public Object call() throws Exception {
					return invoke(obj, methodName, args);
				}
			});
		}
		
		List<Future<Object>> invokeAll = tpe.invokeAll(tasks);
		
		for (Future<Object> future : invokeAll) {
			if(future.isDone()){
				set.add(future.get());
			}
		}
		
		Thread.sleep(1000L);
		
		return set;
	}
	
	/**
	 * 方法调用
	 * @param clazz
	 * @param methodName
	 * @param args
	 * @return
	 * @throws InstantiationException
	 * @throws IllegalAccessException
	 * @throws InvocationTargetException
	 */
	private static <T> Object invoke(final Class<T> clazz, final String methodName, final Object args)
			throws InstantiationException, IllegalAccessException, InvocationTargetException {
		final T newInstance = clazz.newInstance();
		Method method = null;
		try {
			method = clazz.getMethod(methodName, args.getClass());
		} catch (NoSuchMethodException e) {
			try {
				method = clazz.getMethod(methodName, args.getClass().getInterfaces());
			} catch (NoSuchMethodException e1) {
				e1.printStackTrace();
			}
		}
		return method.invoke(newInstance, args);
	}
	
	/**
	 * 方法调用
	 * @param obj
	 * @param methodName
	 * @param args
	 * @return
	 * @throws NoSuchMethodException
	 * @throws IllegalAccessException
	 * @throws InvocationTargetException
	 */
	private static Object invoke(final Object obj, final String methodName, final Object args)
			throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
		Method method = obj.getClass().getMethod(methodName, args.getClass());
		return method.invoke(obj, args);
	}
	
	/**
	  * @Description: 业务提交
	  * @author zhoutao
	  * @date 2017年8月17日 下午2:31:25 
	  * @version V1.0
	 */
	public interface BizSubmit {
		
		Object submit() throws Exception;
	}

}
